package com.zhuchang.self_rpc.test.client;

import com.zhuchang.self_rpc.test.codec.RpcRequest;
import com.zhuchang.self_rpc.test.codec.RpcResponse;
import com.zhuchang.self_rpc.test.common.RpcFuture;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import lombok.Builder;
import lombok.Data;

import java.net.SocketAddress;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @Author RenPu
 * @Date 2021/8/17 17:44
 * @Version 1.0
 * @Description: 基于netty自定义RPC框架：业务执行器的编写
 **/


public class RpcClientHandler extends SimpleChannelInboundHandler<RpcResponse> {


    /**
     * 通道对象
     **/
    private Channel channel;

    /**
     * 连接对象
     **/
    private SocketAddress remotePeer;

    /**
     * key :requestId
     **/
    private Map<String, RpcFuture> pendingRpcTable = new ConcurrentHashMap<String, RpcFuture>();


    /**
     * 通道激活时候，触发的方法
     *
     * @param cxf
     */
    @Override
    public void channelActive(ChannelHandlerContext cxf) throws Exception {
        super.channelActive(cxf);
        this.remotePeer = this.channel.remoteAddress();

    }


    /**
     * 通道注册时候，触发的方法
     *
     * @param cxf
     * @throws Exception
     */
    @Override
    public void channelRegistered(ChannelHandlerContext cxf) throws Exception {
        super.channelRegistered(cxf);
        this.channel = cxf.channel();
    }


    /**
     * @param channelHandlerContext
     * @param response
     * @throws Exception
     */
    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, RpcResponse response) throws Exception {
        String requestId = response.getRequestId();
        RpcFuture rpcFuture = pendingRpcTable.get(requestId);
        if (rpcFuture != null) {
            pendingRpcTable.remove(requestId);
            rpcFuture.done(response);
        }


    }


    /**
     * 获取当前通道的连接对象
     *
     * @return
     */
    public SocketAddress getRemotePeer() {
        return this.remotePeer;
    }


    /**
     * 释放通道资源的方法
     * Netty提供了一种，主动关闭连接的方式。发送Unpooled.EMPTY_BUFFER 监听器就会主动关闭连接
     */
    public void close() {
        channel.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
    }


    /**
     * 异步发送请求方法
     *
     * @param rpcRequest
     * @return
     */
    public RpcFuture sendRequest(RpcRequest rpcRequest) {

        //将rpcRequest封装为future模型
        RpcFuture rpcFuture = new RpcFuture(rpcRequest);

        //
        pendingRpcTable.put(rpcRequest.getRequestId(), rpcFuture);

        //异步将rpcRequest对象写出去
        channel.writeAndFlush(rpcRequest);
        return rpcFuture;
    }


}
